package com.mangocool.kafkaspark;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.PropertyConfigurator;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;

public class kafkaProducer extends Thread  {

	private String topic;
	 
    public kafkaProducer(String topic){
        super();
        this.topic = topic;
    }
 
    @Override
    public void run() {
        Producer producer = createProducer();
        int i=0;
        while(true){
            producer.send(new KeyedMessage<Integer, String>(topic, "message: " + i++));
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
 
    private Producer createProducer() {
        Properties properties = new Properties();
        properties.put("zookeeper.connect", "10.0.104.20:2181,10.0.104.21:2181,10.0.104.22:2181");//声明zk
        properties.put("serializer.class", StringEncoder.class.getName());
        properties.put("metadata.broker.list", "10.0.104.20:9092");// 声明kafka broker
        return new Producer<Integer, String>(new ProducerConfig(properties));
    }
 
    public static void main(String[] args) {
    	System.out.println(System.getProperty("user.dir"));
    	String path = System.getProperty("user.dir");
    	String log4jConfPath = path+"\\src\\main\\resources\\log4j.properties";
    	PropertyConfigurator.configure(log4jConfPath);
        new kafkaProducer("kafka-spark-demo").start();// 使用kafka集群中创建好的主题 kafka-spark-demo
 
    }
}
